package com.bw.realtime.dwd.app;

import com.bw.medical.realtime.common.base.BaseSQL;
import com.bw.medical.realtime.common.constant.Constant;
import com.bw.medical.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeConsultation extends BaseSQL {
    public static void main(String[] args) {
        new DwdTradeConsultation().start(
                10011,
                4,
                "dwd_trade_consultation"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        //从kafka中的topic_db 主题中读取数据 创建动态表
        readOdsDb(tabEnv,"dwd_trade_consultation");

//        tabEnv.sqlQuery("select data from topic_db where `table` = 'consultation' and `type` = 'insert' ").execute().print();

        // 筛选数据
        Table result_table = tabEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['consultation_fee'] consultation_fee,\n" +
                "`data`['doctor_id'] doctor_id,\n" +
                "`data`['patient_id'] patient_id,\n" +
                "`data`['user_id'] user_id\n" +
                "from topic_db\n" +
                "where `database` = 'medical'\n" +
                "and `table` = 'consultation'\n" +
                "and `type` = 'insert'");

        //创建临时表
        tabEnv.createTemporaryView("result_table",result_table);

        //创建动态表 和 要写入的主题进行映射
        tabEnv.executeSql("create table "+Constant.DWD_TRADE_CONSULTATION+" (\n" +
                "`id` string,\n" +
                "`create_time` string,\n" +
                "`consultation_fee` string,\n" +
                "`doctor_id` string,\n" +
                "`patient_id` string,\n" +
                "`user_id` string\n" +
                ")" +  SQLUtil.getKafkaSinkSQL(Constant.DWD_TRADE_CONSULTATION)) ;
        //写入
        result_table.executeInsert(Constant.DWD_TRADE_CONSULTATION);


    }
}
